package com.example.flinkcourse.lesson4.config;

import com.example.flinkcourse.lesson4.model.Event;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.ExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * 执行器配置示例
 * 展示各种执行配置的使用方式
 */
public class ExecutionConfigExample {
    private static final Logger LOG = LoggerFactory.getLogger(ExecutionConfigExample.class);

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 获取执行配置
        ExecutionConfig executionConfig = env.getConfig();
        
        // 1. 配置并行度
        LOG.info("配置并行度...");
        executionConfig.setParallelism(4);  // 设置默认并行度
        executionConfig.setMaxParallelism(16);  // 设置最大并行度
        
        // 2. 配置执行模式
        LOG.info("配置执行模式...");
        executionConfig.setExecutionMode(ExecutionMode.PIPELINED);  // 设置流式执行模式
        
        // 3. 配置对象重用
        LOG.info("配置对象重用...");
        executionConfig.enableObjectReuse();  // 启用对象重用以提高性能
        
        // 4. 配置序列化
        LOG.info("配置序列化...");
        executionConfig.enableForceKryo();  // 强制使用 Kryo 序列化
        executionConfig.registerKryoType(Event.class);  // 注册 Event 类型到 Kryo
        
        // 5. 配置全局参数
        LOG.info("配置全局参数...");
        Map<String, String> globalParams = new HashMap<>();
        globalParams.put("source.type", "kafka");
        globalParams.put("checkpoint.interval", "60000");
        executionConfig.setGlobalJobParameters(new org.apache.flink.configuration.Configuration().fromMap(globalParams));
        
        // 6. 配置任务取消
        LOG.info("配置任务取消...");
        executionConfig.setTaskCancellationInterval(5000);  // 设置任务取消间隔为 5 秒
        
        // 7. 配置重启策略
        LOG.info("配置重启策略...");
        env.setRestartStrategy(
            RestartStrategies.fixedDelayRestart(
                3,  // 最大重启次数
                Time.of(10, TimeUnit.SECONDS)  // 重启延迟
            )
        );
        
        // 8. 配置检查点
        LOG.info("配置检查点...");
        env.enableCheckpointing(60000);  // 每 60 秒进行一次检查点
        env.getCheckpointConfig().setCheckpointTimeout(30000);  // 检查点超时时间
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);  // 最大并发检查点数量
        
        // 9. 配置状态后端
        LOG.info("配置状态后端...");
        env.setStateBackend(new org.apache.flink.contrib.streaming.state.RocksDBStateBackend(
            "hdfs:///flink/state",
            true  // 启用增量检查点
        ));
        
        // 10. 配置水印
        LOG.info("配置水印...");
        env.getConfig().setAutoWatermarkInterval(200);  // 设置水印生成间隔为 200 毫秒
        
        // 创建示例数据流
        env.addSource(new org.apache.flink.streaming.api.functions.source.SourceFunction<Event>() {
            private volatile boolean running = true;
            private long count = 0;
            
            @Override
            public void run(SourceContext<Event> ctx) throws Exception {
                while (running) {
                    Event event = Event.builder()
                        .id("event-" + count)
                        .type("test")
                        .timestamp(System.currentTimeMillis())
                        .amount(Math.random() * 100)
                        .build();
                    
                    ctx.collect(event);
                    count++;
                    
                    Thread.sleep(1000);
                }
            }
            
            @Override
            public void cancel() {
                running = false;
            }
        })
        .setParallelism(2)  // 设置数据源并行度
        .print()  // 打印输出
        .setParallelism(1);  // 设置输出并行度
        
        // 执行作业
        env.execute("Execution Config Example");
    }
} 